package cn.hesion.mrreducejoin;
/**
 * ClassName: ReduceJoinReducer <br/>
 * Description: <br/>
 * date: 2021/2/3 11:57<br/>
 *
 * @author Hesion<br />
 * @version
 * @since JDK 1.8
 */

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;

/**
 * @program: ClientDemo
 * @description:
 * @author: hesion
 * @create: 2021-02-03 11:57
 **/
public class ReduceJoinReducer extends Reducer<Text,DeliverBean,DeliverBean, NullWritable> {
    @Override
    protected void reduce(Text key, Iterable<DeliverBean> values, Context context) throws IOException, InterruptedException {
        // 准用于收集投递信息的对象数组
        ArrayList<DeliverBean> deBeans = new ArrayList<>();
        // 用于接收position的对象
        DeliverBean pBean = new DeliverBean();

        for (DeliverBean bean : values) {
            if("deliver".equals(bean.getFlag())){
                // 表示这是投递信息的bean
                DeliverBean dBean = new DeliverBean();
                try {
                    BeanUtils.copyProperties(dBean,bean);
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                } catch (InvocationTargetException e) {
                    e.printStackTrace();
                }
                deBeans.add(dBean);
            }else {
                try {
                    BeanUtils.copyProperties(pBean,bean);
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                } catch (InvocationTargetException e) {
                    e.printStackTrace();
                }
            }
        }

        for (DeliverBean bean : deBeans) {
            bean.setPositionName(pBean.getPositionName());
            context.write(bean,NullWritable.get());
        }


    }
}
